package requestCase

import org.apache.spark.{SparkConf, SparkContext}

object Req01_HotCategoryTop10Analysis {
  def main(args: Array[String]): Unit = {

    val req01_HotCategoryTop10AnalysisConf = new SparkConf().setMaster("local[*]").setAppName("Req01_HotCategoryTop10Analysis")
    val sparkContext = new SparkContext(req01_HotCategoryTop10AnalysisConf)

    val logs = sparkContext.textFile("src\\main\\scala\\data\\user_visit_action.txt")
    logs.cache()

    val clickRDD = logs.filter(
      x => {
        val dates = x.split("_")
        dates(6) != "-1"
      }
    )
    val clickRDDcount = clickRDD.map(
      date => {
        val strings = date.split("_")
        (strings(6), 1)
      }
    ).reduceByKey(_+_)

    val orderRDD = logs.filter(
      x => {
        val dates = x.split("_")
        dates(8) != "null"
      }
    )
    val orderRDDcount = orderRDD.flatMap(
      date => {
        val strings = date.split("_")
        val strings1 = strings(8).split(",")
        strings1.map(
          x => (x, 1)
        )
      }
    ).reduceByKey(_+_)

    val payRDD = logs.filter(
      dates => {
        val strings = dates.split("_")
        strings(10) != "null"
      }
    )

    val payRDDcount = payRDD.flatMap(
      date => {
        val strings = date.split("_")
        val payID = strings(10).split(",")
        payID.map(x => (x, 1))
      }
    ).reduceByKey(_ + _)

    val cogroupRDD = clickRDDcount.cogroup(orderRDDcount, payRDDcount)

    val resultRDD = cogroupRDD.mapValues {
      case (click, order, pay) => {

        var clickCnt = 0
        val click_iterator = click.iterator
        if (click_iterator.hasNext) {
          clickCnt = click_iterator.next()
        }

        var orderCnt = 0
        val order_iterator = order.iterator
        if (order_iterator.hasNext) {
          orderCnt = order_iterator.next()
        }

        var payCnt = 0
        val pay_iterator = pay.iterator
        if (pay_iterator.hasNext) {
          payCnt = pay_iterator.next()
        }

        (clickCnt, orderCnt, payCnt)
      }
    }

    val result_tuples = resultRDD.sortBy(_._2, false).take(10)

    result_tuples.foreach(println)

    sparkContext.stop()
  }

}
